package com.weimob.mcloud.wechatwork.customer.mq;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.google.common.collect.Lists;
import com.weimob.artemis.client.consumer.model.ConsumerRecord;
import com.weimob.artemis.client.consumer.model.MessageResult;
import com.weimob.artemis.client.spring.annotation.ArtemisConsumerListener;
import com.weimob.artemis.client.spring.annotation.Subscribe;
import com.weimob.mcloud.wechatwork.customer.domain.ArtemisConstant;
import com.weimob.mcloud.wechatwork.customer.domain.ExternalUserChangeInfo;
import com.weimob.mcloud.wechatwork.customer.domain.enums.CorpExternaluserChangEventEnum;
import com.weimob.mcloud.wechatwork.customer.mq.dto.AddExternalUserSecondDTO;
import com.weimob.mcloud.wechatwork.customer.mq.handler.externaluser.AddCorpExternaluserSecondHandler;
import com.weimob.mcloud.wechatwork.customer.mq.handler.externaluser.CorpExternalUserHandler;
import com.weimob.mcloud.wechatwork.customer.service.user.enums.YesOrNoEnum;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;

import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

@Component
@Slf4j
public class CorpExternalUserConsumer {

    @Autowired
    private Map<String, CorpExternalUserHandler> corpExternaluserHandlerMap;


    @Autowired
    private AddCorpExternaluserSecondHandler addCorpExternaluserSecondHandler;


    @Autowired
    @Qualifier("changeCorpExternaluserThreadPool")
    private ThreadPoolTaskExecutor changeCorpExternaluserThreadPool;

    @Autowired
    @Qualifier("secondCorpExternaluserThreadPool")
    private ThreadPoolTaskExecutor secondCorpExternaluserThreadPool;

    @Autowired
    @Qualifier("historySecondCorpExternaluserThreadPool")
    private ThreadPoolTaskExecutor historySecondCorpExternaluserThreadPool;


    @Value("${externaluser.second.partitioncount:25}")
    private Integer sendPartitionCount;

    @Value("${externaluser.sleep.interval:3000}")
    private Integer sleepInterval;


    @Value("${externaluser.sleep.flag:true}")
    private Boolean sleepFlag;


    public Map<String, List<ExternalUserChangeInfo>> classifyExternalUserChangeInfo(List<ExternalUserChangeInfo> externalUserChangeInfos) {
        Map<String, List<ExternalUserChangeInfo>> resultMap = new HashMap<>();
        Map<String, List<ExternalUserChangeInfo>> eventMap = externalUserChangeInfos.stream().filter(item -> Objects.nonNull(item.getEvent())).collect(Collectors.groupingBy(ExternalUserChangeInfo::getEvent));

        eventMap.forEach((k, v) -> {
            if (k.equals(CorpExternaluserChangEventEnum.ADD.getEvent())) {
                resultMap.putAll(v.stream().collect(Collectors.groupingBy(ExternalUserChangeInfo::getExternalUserID)));
            }
            if (k.equals(CorpExternaluserChangEventEnum.CREATE_TAG.getEvent())) {
                resultMap.putAll(v.stream().collect(Collectors.groupingBy(ExternalUserChangeInfo::getId)));
            }
            if (k.equals(CorpExternaluserChangEventEnum.CREATE_CHAT_GROUP.getEvent())) {
                resultMap.putAll(v.stream().collect(Collectors.groupingBy(ExternalUserChangeInfo::getChatId)));
            }
        });
        return resultMap;
    }

    public void executeExternalUserChangeInfos(List<ExternalUserChangeInfo> list) {
        if (CollectionUtils.isEmpty(list)) {
            log.info("ExternalUserChangeInfo list is empty");
            return;
        }


        // 标签或标签组排序消息会出现无tagType的消息，需要过滤
        List<ExternalUserChangeInfo> userChangeInfos = filterTagTypeIsNull(list);
        if (CollectionUtils.isEmpty(userChangeInfos)) {
            return;
        }


        for (ExternalUserChangeInfo externalUserChangeInfo : userChangeInfos) {

            try {

                //实时消息过来的数据, 非历史数据同步
                externalUserChangeInfo.setHistoryCustomerSync(YesOrNoEnum.NO.getCode());

                String className = CorpExternaluserChangEventEnum
                        .getClassNameByTypeExcludeAdd(externalUserChangeInfo.getEvent(), externalUserChangeInfo.getChangeType());

                CorpExternalUserHandler corpExternaluserHandler = corpExternaluserHandlerMap.get(className);
                if (null == corpExternaluserHandler) {
                    continue;
                }

                corpExternaluserHandler.handleCorpExternalUser(externalUserChangeInfo);

            } catch (Exception e) {
                log.error("Fail to executeExternalUserChangeInfos,args is {},error:", externalUserChangeInfo, e);
            }

        }

    }

    private List<ExternalUserChangeInfo> filterTagTypeIsNull(List<ExternalUserChangeInfo> list) {
        if (CollectionUtils.isEmpty(list)) {
            return new ArrayList<>();
        }

        List<ExternalUserChangeInfo> filter = new ArrayList<>();
        for (ExternalUserChangeInfo externalUserChangeInfo : list) {
            if (null == externalUserChangeInfo) {
                continue;
            }


            if (externalUserChangeInfo.getEvent().equals(CorpExternaluserChangEventEnum.CREATE_TAG.getEvent())) {
                if (StringUtils.isBlank(externalUserChangeInfo.getTagType())) {
                    continue;
                }
            }


            filter.add(externalUserChangeInfo);
        }


        return filter;
    }

    /**
     * 企业微信外部联系人.标签变更消费(编辑、删除).客户群（创建，变更，解散） https://work.weixin.qq.com/api/doc/90001/90143/92277#%E4%BC%81%E4%B8%9A%E5%AE%A2%E6%88%B7%E6%A0%87%E7%AD%BE%E5%88%9B%E5%BB%BA%E4%BA%8B%E4%BB%B6
     */
    @ArtemisConsumerListener(
            value = @Subscribe(topic = ArtemisConstant.Topic.WECHATWORK_SELF_MSG, user = ArtemisConstant.USER), interval = 50)
    public void changeCorpExternaluser(MessageResult result) {

        try {

            log.info("企业微信外部联系人变更消息, body:{}", JSON.toJSONString(result.getRecords()));
            List<ExternalUserChangeInfo> externalUserChangeInfos = covertToCustomerChangeInfoList(result.getRecords());


            // 分类,主要为了保证顺序
            Map<String, List<ExternalUserChangeInfo>> resultMap = classifyExternalUserChangeInfo(externalUserChangeInfos);


            if (resultMap.size() <= 0) {
                log.warn("消费实时外部联系人 >> resultMap is empty error ");
                return;
            }


            CountDownLatch countDownLatch = new CountDownLatch(resultMap.size());
            for (Map.Entry<String, List<ExternalUserChangeInfo>> entry : resultMap.entrySet()) {

                changeCorpExternaluserThreadPool.execute(() -> {
                    try {
                        executeExternalUserChangeInfos(entry.getValue());
                    } catch (Exception e) {
                        log.error("消费实时外部联系人消息变更失败, executeExternalUserChangeInfos error", e);
                    } finally {
                        countDownLatch.countDown();
                    }
                });

            }

            boolean await = countDownLatch.await(5, TimeUnit.SECONDS);
            if (Boolean.FALSE.equals(await)) {
                log.error("消费实时外部联系人消息变更超时");
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            log.error("InterruptedException e", e);
        } catch (Exception e) {
            log.error("消费实时外部联系人Artemis数据处理失败,error", e);
        }
    }


    private List<ExternalUserChangeInfo> covertToCustomerChangeInfoList(List<ConsumerRecord> records) {
        if (CollectionUtils.isEmpty(records)) {
            return new ArrayList<>();
        }

        List<ExternalUserChangeInfo> userChangeInfos = new ArrayList<>();
        for (ConsumerRecord consumerRecord : records) {

            ExternalUserChangeInfo externalUserChangeInfo = JSONObject.
                    parseObject(consumerRecord.getValue(), ExternalUserChangeInfo.class);

            if (null != externalUserChangeInfo) {
                userChangeInfos.add(externalUserChangeInfo);
            }

        }


        return userChangeInfos;
    }

    /**
     * 两阶段处理 实时数据
     */
    @ArtemisConsumerListener(
            value = @Subscribe(topic = ArtemisConstant.Topic.CORP_EXTERNALUSER_SECOND_EVENT, user = ArtemisConstant.CONSUMER_USER), interval = 50)
    public void handleExternalUserSecond(MessageResult result) {

        handleSecond(result, secondCorpExternaluserThreadPool);
    }

    private void handleSecond(MessageResult result, ThreadPoolTaskExecutor secondCorpExternaluserThreadPool) {
        log.info("新增企业微信外部联系人二次分发消息, body:{}", JSON.toJSONString(result));
        List<ConsumerRecord> records = result.getRecords();
        if (CollectionUtils.isEmpty(records)) {
            return;
        }
        try {
            List<AddExternalUserSecondDTO> addExternalUserSecondDTOList = records
                    .stream()
                    .map(record -> JSONObject.parseObject(record.getValue().toString(), AddExternalUserSecondDTO.class))
                    .collect(Collectors.toList());
            if (CollectionUtils.isEmpty(addExternalUserSecondDTOList)) {
                log.error("二阶段外部联系人,result is empty error ");
                return;
            }

            List<List<AddExternalUserSecondDTO>> partition = Lists.partition(addExternalUserSecondDTOList, sendPartitionCount);

            for (List<AddExternalUserSecondDTO> addExternalUserSecondDTOS : partition) {
                CompletableFuture<Object>[] completableFutures = new CompletableFuture[addExternalUserSecondDTOS.size()];
                for (int i = 0; i < addExternalUserSecondDTOS.size(); i++) {
                    AddExternalUserSecondDTO addExternalUserSecondDTO = addExternalUserSecondDTOS.get(i);
                    completableFutures[i] = CompletableFuture.supplyAsync(() -> {
                        addCorpExternaluserSecondHandler.handleCorpExternalUser(addExternalUserSecondDTO);
                        return null;
                    }, secondCorpExternaluserThreadPool);
                }
                //等待执行完成
                CompletableFuture.allOf(completableFutures).join();
            }

        } catch (Exception e) {
            log.error("消费实时外部联系人Artemis数据处理失败. message:{} ", JSON.toJSONString(records), e);
        }
    }

    /**
     * 两阶段处理历史数据
     *
     * @param result
     */
    @ArtemisConsumerListener(
            value = @Subscribe(topic = ArtemisConstant.Topic.HISTORY_CORP_EXTERNALUSER_SECOND_EVENT, user = ArtemisConstant.CONSUMER_USER), pollSize = 50, interval = 50)
    public void handleHistoryExternalUserSecond(MessageResult result) {
        if (Boolean.TRUE.equals(sleepFlag)) {
            try {
                Thread.sleep(sleepInterval);
            } catch (InterruptedException e) {
                log.error("历史客户同步，sleep报错：", e);
                Thread.currentThread().interrupt();
            }
        }

        handleSecond(result, historySecondCorpExternaluserThreadPool);
    }


}
